/*
* Change Logs:
* Date           Author       Notes
* 2020-10-16     chenbin      mqtt client base on paho
*/
#include <string.h>
#include <rtthread.h>
#include "small_mqtt_client.h"

#ifdef MQTT_CLIENT_RTTHREAD_PORT_SOCKET

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <ctype.h>

//#if defined(RT_USING_SAL)
//#include <sal.h>
//#include <sal_netdb.h>
//#include <sal_socket.h>
//#endif
#if defined(SAL_USING_POSIX)
#include <netdb.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <sys/errno.h>
#include <sys/ioctl.h>
#endif

static int _mqtt_rtsocket_connect(mqtt_client_t *client)
{
	int ret = MQTT_FAIL;
	unsigned long mode = 0;
	struct addrinfo hints, *addr_list, *cur;
	struct timeval timeout;
	
	struct sockaddr_in * socket_addr;
  char addr_str[64];

	client->socket_fd = -1;
	client->_next_packetid = 0;
	
	/* Do name resolution with both IPv6 and IPv4 */
	memset(&hints, 0, sizeof(hints));
	hints.ai_family = AF_INET;
	hints.ai_socktype = SOCK_STREAM;
	hints.ai_protocol = IPPROTO_TCP;
	//hints.ai_protocol = PROTOCOL_TLS;

	ret = getaddrinfo(client->chost, client->cport, &hints, &addr_list);
	if (ret != 0)
	{
			rt_kprintf("mqtt socket getaddrinfo fail [%d]\n",ret);
			return ret;
	}
	for (cur = addr_list; cur != NULL; cur = cur->ai_next)
	{

#ifdef SAL_USING_TLS
//				if(c->is_tls)
//				{
//					c->socketfd = socket(cur->ai_family,SOCK_STREAM, PROTOCOL_TLS);
//				}
//				else
#endif
			{
				client->socket_fd = socket(cur->ai_family,SOCK_STREAM, IPPROTO_TCP);
			}
			if (client->socket_fd < 0) 
			{
					rt_kprintf("mqtt socket[%d] < 0\n", client->socket_fd);
					rt_thread_mdelay(100);
					continue;
			}
			mode = 0;
			ioctlsocket(client->socket_fd, FIONBIO, &mode);

			ret = connect(client->socket_fd, cur->ai_addr, cur->ai_addrlen);
			
			socket_addr = (struct sockaddr_in *)(cur->ai_addr);
			
			inet_ntoa_r(socket_addr->sin_addr.s_addr, addr_str, sizeof(addr_str) - 1);
			
			rt_kprintf("mqtt socket[%d] connect: %s:%d  ret:%d\n", client->socket_fd, addr_str ,ntohs(socket_addr->sin_port),ret);
			if (ret == 0)
			{
					break;//连接成功
			}
			closesocket(client->socket_fd);
			rt_thread_mdelay(100);
	}
	freeaddrinfo(addr_list);
	
	mode = 0;
	ioctlsocket(client->socket_fd, FIONBIO, &mode);  //block recv
	
	timeout.tv_sec = client->read_timeout/1000;
	timeout.tv_usec = (client->read_timeout%1000)*1000;
	setsockopt(client->socket_fd, SOL_SOCKET, SO_RCVTIMEO, (void *) &timeout,sizeof(timeout));  //recv timeout

	timeout.tv_sec = client->write_timeout/1000;
	timeout.tv_usec = (client->write_timeout%1000)*1000;
	setsockopt(client->socket_fd, SOL_SOCKET, SO_SNDTIMEO, (void *) &timeout,sizeof(timeout));  //send timeout

	ret = MQTT_OK;
	return ret;
}

static int _mqtt_rtsocket_disconnect(mqtt_client_t *client)
{
	//mqtt_client_port_socket_t *port_socket = (mqtt_client_port_socket_t *)client->port;
	
	if (client->socket_fd >= 0) 
	{
			closesocket(client->socket_fd);
	}
	return MQTT_OK;
}
/*
 */
static int _mqtt_rtsocket_recv(mqtt_client_t *client, unsigned char *buf, int len, int timeout_ms)
{
	int rc = 0;
	int socket_status = 0;
	socklen_t slen = 0;
	struct timeval timeout = {0};
	
	timeout.tv_sec = timeout_ms / 1000;
	timeout.tv_usec = (timeout_ms % 1000)*1000;
	
	if( (timeout.tv_sec == 0 && timeout.tv_usec <= 0))
	{
			timeout.tv_sec = 0;
			timeout.tv_usec = 1000*100;
	}
	setsockopt(client->socket_fd, SOL_SOCKET, SO_RCVTIMEO, (void *) &timeout,sizeof(timeout));
	
	rc = recv(client->socket_fd, buf, len, 0);
	if(rc == 0) //socket wait_close
	{
		rt_kprintf("mqtt socket[%d] recv rc:%d wait_close\n",client->socket_fd, rc);
		rc = MQTT_RESTART;
	}else
	if(rc < 0) //socket error
	{
//		if(!(errno == EINTR || errno == EWOULDBLOCK || errno == EAGAIN))
//		{
//					rt_kprintf("mqtt socket[%d] recv rc:%d err:%d \n",client->socket_fd, rc, errno);
//				rc = MQTT_RESTART;
//		}else
//		{
//				//rt_kprintf("rtsocket timeout :%d\n",rc);
//				rc = MQTT_TIMEOUT;
//		}
		slen = sizeof(socket_status);
		getsockopt(client->socket_fd, SOL_SOCKET, SO_ERROR, (void *)&socket_status , &slen );
		
		if((socket_status == 0 || socket_status == EINTR || socket_status == EWOULDBLOCK || socket_status == EAGAIN))
		{
				//rt_kprintf("rtsocket timeout :%d\n",rc);
				rc = MQTT_TIMEOUT;
		}else
		{
				rt_kprintf("mqtt socket[%d] recv rc:%d err:%d \n",client->socket_fd, rc, socket_status);
				rc = MQTT_RESTART;
		}
	}
	return rc;
}

static int _mqtt_rtsocket_send(mqtt_client_t *client, unsigned char *buf, int len)
{
	int rc = 0;
	
	rc = send(client->socket_fd, buf, len, 0);
	if(rc < 0)
	{
		rt_kprintf("mqtt socket[%d] send rc:%d \n",client->socket_fd, rc);
		rc = MQTT_RESTART;
	}else
	{
		rc = MQTT_OK;
	}
	return rc;
}

int mqtt_port_socket_init(mqtt_client_port_socket_t *port)
{
	*(uint32_t *)(&port->base.magic) = MQTT_PORT_SOCKET;
	port->base.net_connect = _mqtt_rtsocket_connect;
	port->base.net_disconnect = _mqtt_rtsocket_disconnect;
	port->base.net_recv = _mqtt_rtsocket_recv;
	port->base.net_send = _mqtt_rtsocket_send;
	return 0;
}

mqtt_client_port_socket_t *mqtt_port_socket_create(void)
{
	mqtt_client_port_socket_t *port_socket = rt_malloc_align(sizeof(mqtt_client_port_socket_t),4);
	if(port_socket)
	{
		rt_memset(port_socket,0,sizeof(mqtt_client_port_socket_t));
		mqtt_port_socket_init(port_socket);
		return port_socket;
	}
	return NULL;
}

mqtt_client_port_socket_t *mqtt_port_socket_get(mqtt_client_t *client)
{
	if(client->port->net_connect == _mqtt_rtsocket_connect)
	{
		return (mqtt_client_port_socket_t*)client->port;
	}
	return NULL;
}

#endif
